﻿// ———————————————————————–
// <copyright file="DEProcessing.cs" company="The MITRE Corporation">
//    Copyright (c) 2010 The MITRE Corporation. All rights reserved.
//    Licensed under the Apache License, Version 2.0 (the "License");
//    you may not use this file except in compliance with the License.
//    You may obtain a copy of the License at
//    http://www.apache.org/licenses/LICENSE-2.0
//    Unless required by applicable law or agreed to in writing, software
//    distributed under the License is distributed on an "AS IS" BASIS,
//    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//    See the License for the specific language governing permissions and
//    limitations under the License.
// </copyright>
// ———————————————————————–
/////////////////////////////////////////////////////////////////////////////////////////////////
// DEProcessing.cs - Service implementation that pulls DE messages from a SQS queue and writes
//                   them to DynamoDB
// Project: EDXLSharp_AWSRouter- DEProcessing
//
// Language:    C#, .NET 4.0
// Platform:    Windows 7, VS 2013
// Author:      Don McGarry The MITRE Corporation
/////////////////////////////////////////////////////////////////////////////////////////////////

// Copyright (c) 2010 The MITRE Corporation. All rights reserved.
//
// NOTICE
// This software was produced for the U. S. Government
// under Contract No. FA8721-09-C-0001, and is
// subject to the Rights in Noncommercial Computer Software
// and Noncommercial Computer Software Documentation Clause
// (DFARS) 252.227-7014 (JUN 1995)

using Amazon.DynamoDBv2;
using Amazon.DynamoDBv2.Model;
using Amazon.SQS;
using Amazon.SQS.Model;
using Amazon.Util;
using EDXLSharp.EDXLDELib;
using System;
using System.Collections.Generic;
using System.ServiceProcess;
using System.Threading;
using ICNETServices;
using System.Globalization;
// Configure log4net using the .config file
[assembly: log4net.Config.XmlConfigurator(Watch = true)]

namespace DEProcessing
{
  /// <summary>
  /// Class for the DE Processing Windows Service
  /// </summary>
  public partial class DEProcessing : ServiceBase
  {
    /// <summary>
    /// Log4net logging object
    /// </summary>
    private static readonly log4net.ILog Log = log4net.LogManager.GetLogger(System.Reflection.MethodBase.GetCurrentMethod().DeclaringType);

    /// <summary>
    /// Thread to run the SQS polling on
    /// </summary>
    private Thread runThread;

    /// <summary>
    /// Amazon SQS configuration object 
    /// </summary>
    private AmazonSQSConfig amazonSQSConfig;

    /// <summary>
    /// Amazon SQS Client
    /// </summary>
    private AmazonSQSClient amazonSQSClient;

    /// <summary>
    /// Amazon DynamoDB configuration object
    /// </summary>
    private AmazonDynamoDBConfig amazonDynamoConfig;

    /// <summary>
    /// Amazon DynamoDB client
    /// </summary>
    private AmazonDynamoDBClient amazonDynamoClient;

    /// <summary>
    /// Initializes a new instance of the DEProcessing class
    /// </summary>
    public DEProcessing()
    {
      this.InitializeComponent();
      this.ServiceName = "DEProcessing Service";
    }

    /// <summary>
    /// Method called when service is started
    /// </summary>
    /// <param name="args">Command Arguments</param>
    protected override void OnStart(string[] args)
    {
      Log.Info("OnStart");
      try
      {
        this.amazonSQSConfig = new AmazonSQSConfig();
        this.amazonSQSConfig.ServiceURL = AWSConstants.SQSServiceURL;
       
        this.amazonSQSClient = new AmazonSQSClient(this.amazonSQSConfig);
        this.amazonDynamoConfig = new AmazonDynamoDBConfig();
        this.amazonDynamoConfig.ServiceURL = AWSConstants.DynamoServiceURL;

        this.amazonDynamoClient = new AmazonDynamoDBClient(this.amazonDynamoConfig);
        this.runThread = new Thread(new ThreadStart(this.PollSQS));
        this.runThread.IsBackground = true;
        this.runThread.Start();
        Log.Info("OnStartComplete");
      }
      catch (Exception e)
      {
        Log.Error("OnStartFailed", e);
      }
      
    }

    /// <summary>
    /// Method called when service is stopped
    /// </summary>
    protected override void OnStop()
    {
      Log.Info("OnStop");
      if (this.runThread != null && this.runThread.IsAlive)
      {
        Log.Info("KillingThread");
        this.runThread.Abort();
        Log.Info("ThreadKilled");
      }
      else
      {
        Log.Warn("Poller thread was not running.");
      }
      Log.Info("OnStopComplete");
    }

    /// <summary>
    /// Thread worker method
    /// </summary>
    private void PollSQS()
    {
      Log.Info("InThreadSQS");
      while (true)
      {
        Log.Info("SQSLoop");
        ReceiveMessageRequest req = new ReceiveMessageRequest();
        req.QueueUrl = AWSConstants.DEProcessingQueueUrl;
        req.WaitTimeSeconds = AWSConstants.DEProcessing.PollIntervalInSeconds;

        ReceiveMessageResponse resp = this.amazonSQSClient.ReceiveMessage(req);
        Log.Info("Got " + resp.Messages.Count + " messages from SQS");
        foreach (Message m in resp.Messages)
        {
          try
          {
            Log.Info(m.Body);
            EDXLDE de = new EDXLDE(m.Body);
            switch (de.DistributionType)
            {
              case TypeValue.Cancel:
                this.ExpireDE(de);
                break;
              case TypeValue.Report:
              case TypeValue.Update:
                if (this.IsLatestDE(de.SenderID, de.DistributionID, de.DateTimeSent))
                {
                  this.AddDEToCache(de, m.Body);
                }
                else
                {
                  Log.Info("DE received was already received or older than a previously received DE.");
                }

                break;
              default:
                Log.Error("Got unexpected distributiontype");
                break;
            }

            
            DeleteMessageRequest deleteMessageRequest = new DeleteMessageRequest();
            deleteMessageRequest.QueueUrl = req.QueueUrl;
            deleteMessageRequest.ReceiptHandle = m.ReceiptHandle;
            try
            {
              DeleteMessageResponse sqsdelresp = this.amazonSQSClient.DeleteMessage(deleteMessageRequest);
              if (sqsdelresp.HttpStatusCode == System.Net.HttpStatusCode.OK)
              {
                Log.Info("Message removed from queue");
              }
              else
              {
                Log.Error("SQS Delete error: " + sqsdelresp.HttpStatusCode.ToString());
              }
            }
            catch (Exception e)
            {
              Log.Error("Error deleting from SQS", e);
            }
          }
          catch (Exception e)
          {
            Log.Error(e);
          }
        }
      }
    }

    /// <summary>
    /// Adds the DE object to the Delete queue for processing
    /// </summary>
    /// <param name="body">DE object</param>
    private void AddDEToDeleteQueue(string body)
    {
      //if de is a report or update, try to send to Content Feed Processing queue
      SendMessageResponse cfResp = amazonSQSClient.SendMessage(AWSConstants.DeleteProcessingQueueUrl, body);
      if (cfResp.HttpStatusCode == System.Net.HttpStatusCode.OK)
      {
        Log.Info("Message sent to SQS Queue: " + AWSConstants.DeleteProcessingQueueUrl);
      }
      else
      {
        Log.Error("SQS Error for queue: " + AWSConstants.DeleteProcessingQueueUrl + cfResp.HttpStatusCode.ToString());
      }
    }

    /// <summary>
    /// Expires the DE in dynamo if it exists. Expiring sets the _ExpiresTime to now minus one second.
    /// </summary>
    /// <param name="de"></param>
    private void ExpireDE(EDXLDE de)
    {
      UpdateItemRequest updateReq = new UpdateItemRequest()
      {
        TableName = AWSConstants.DEProcessing.DynamoInfo.TableName,
        Key = new Dictionary<string, AttributeValue>(),
        AttributeUpdates = new Dictionary<string,AttributeValueUpdate>()
      };

      updateReq.Key.Add(AWSConstants.DEProcessing.DynamoInfo.HashName, new AttributeValue() { S = de.DistributionID });
      updateReq.Key.Add(AWSConstants.DEProcessing.DynamoInfo.RangeName, new AttributeValue() { S = de.SenderID });

      updateReq.UpdateExpression = "SET #et = :expTime";
      updateReq.ExpressionAttributeNames = new Dictionary<string, string>
      {
        {"#et", "_ExpiresTime"}
      };

      updateReq.ExpressionAttributeValues = new Dictionary<string, AttributeValue>
      {
        {":expTime", new AttributeValue{S = DateTime.Now.Subtract(new TimeSpan(0,0,1)).ToUniversalTime().ToString(AWSSDKUtils.ISO8601DateFormat)}}
      };


      updateReq.ConditionExpression = "attribute_exists(#et)";
      
      
     
      try
      {
        UpdateItemResponse updateResp = this.amazonDynamoClient.UpdateItem(updateReq);
        if (updateResp.HttpStatusCode == System.Net.HttpStatusCode.OK)
        {
          Log.Info("DE expired in dynamo");
        }
        else
        {
          Log.Error("Dynamo Error" + updateResp.HttpStatusCode.ToString());
        }
      }
      catch (ConditionalCheckFailedException)
      {
        Log.Info("Coudldn't expire DE because it doesn't exist in DB.");
      }
      catch (Exception e)
      {
        Log.Error("Error expiring DE in dynamo", e);
      }
    }

    /// <summary>
    /// Adds an EDXL-DE Object to Dynamo
    /// </summary>
    /// <param name="de">The DE Object</param>
    /// <param name="body">The Serialized DE Object Sting</param>
    private void AddDEToCache(EDXLDE de, string body)
    {
      PutItemRequest dynamoreq = new PutItemRequest();
      Dictionary<string, AttributeValue> dynamoitem = new Dictionary<string, AttributeValue>();
      
      //add DE ID as the hash key
      AttributeValue dynamovalue = new AttributeValue();
      dynamovalue.S = de.DistributionID;
      dynamoitem.Add(AWSConstants.DEProcessing.DynamoInfo.HashName, dynamovalue);
      
      //add Sender ID as the range key
      dynamovalue = new AttributeValue();
      dynamovalue.S = de.SenderID;
      dynamoitem.Add(AWSConstants.DEProcessing.DynamoInfo.RangeName, dynamovalue);

      //TODO: create method to read DynamoInfo values (fields) and set the value through code
      //add Date Time Sent
      dynamovalue = new AttributeValue();
      dynamovalue.S = de.DateTimeSent.ToUniversalTime().ToString(AWSSDKUtils.ISO8601DateFormat);
      dynamoitem.Add("dateTimeSent", dynamovalue);

      //add the DE XML
      dynamovalue = new AttributeValue();
      dynamovalue.S = body;
      dynamoitem.Add("EDXLDistribution", dynamovalue);
      dynamoreq.Item = dynamoitem;

      //add the expires time
      dynamovalue = new AttributeValue();
      dynamovalue.S = this.DetermineExpiresTime(de).ToUniversalTime().ToString(AWSSDKUtils.ISO8601DateFormat);
      dynamoitem.Add("_ExpiresTime", dynamovalue);
      dynamoreq.Item = dynamoitem;

      dynamoreq.TableName = AWSConstants.DEProcessing.DynamoInfo.TableName;
      try
      {
        PutItemResponse dynamoresp = this.amazonDynamoClient.PutItem(dynamoreq);
        if (dynamoresp.HttpStatusCode == System.Net.HttpStatusCode.OK)
        {
          Log.Info("Message Added to Dynamo");
        }
        else
        {
          Log.Error("Dynamo Error" + dynamoresp.HttpStatusCode.ToString());
        }
      }
      catch (Exception e)
      {
        Log.Error("Error writing DE to dynamo", e);
      }
    }

    /// <summary>
    /// Determines the overall expiration time of a DE. This looks at the ContentObjects' ExpiresTimes.
    /// If none of them are set, 1 day from the DE time sent is used.
    /// </summary>
    /// <param name="de">The DE</param>
    /// <returns>The expiration time.</returns>
    private DateTime DetermineExpiresTime(EDXLDE de)
    {
      if(de.ExpiresDateTime.HasValue)
      {
        return de.ExpiresDateTime.Value;
      }
      else
      {
        return de.DateTimeSent.AddDays(1);
      }
    }

    /// <summary>
    /// Marks the DE the specified SenderID and DistributionID for deletion from DynamoDB
    /// </summary>
    /// <param name="senderID">The DE SenderID</param>
    /// <param name="distributionID">The DE DistributionID</param>
    private void MarkDEForDeletion(string senderID, string distributionID)
    {
      UpdateItemRequest dynamoreq = new UpdateItemRequest();
      dynamoreq.TableName = AWSConstants.DEProcessing.DynamoInfo.TableName;

      Dictionary<string, AttributeValue> keys = new Dictionary<string, AttributeValue>();
      
      //add the hash key
      AttributeValue dynamovalue = new AttributeValue();
      dynamovalue.S = distributionID;
      keys.Add(AWSConstants.DEProcessing.DynamoInfo.HashName, dynamovalue);

      //add the range key
      dynamovalue = new AttributeValue();
      dynamovalue.S = senderID;
      keys.Add(AWSConstants.DEProcessing.DynamoInfo.RangeName, dynamovalue);

      //set the keys
      dynamoreq.Key = keys;

      Dictionary<string, AttributeValue> expressionValues = new Dictionary<string, AttributeValue>();
      //add the delete value of "true"
      AttributeValue dynamovalue2 = new AttributeValue();
      dynamovalue2.BOOL = true;
      expressionValues.Add(":newDelete", dynamovalue2);

      //add the expression values
      dynamoreq.ExpressionAttributeValues = expressionValues;

      //set the update query
      dynamoreq.UpdateExpression = "SET _Delete = :newDelete";

      try
      {
        UpdateItemResponse dynamoresp = this.amazonDynamoClient.UpdateItem(dynamoreq);
        if (dynamoresp.HttpStatusCode == System.Net.HttpStatusCode.OK)
        {
          Log.Info("Message Added to Dynamo");
        }
        else
        {
          Log.Error("Dynamo Error" + dynamoresp.HttpStatusCode.ToString());
        }
      }
      catch (Exception e)
      {
        Log.Error("Error writing DE to dynamo", e);
      }
    }

    /// <summary>
    /// Deletes All DE Objects with the specified SenderID and DistributionID from DynamoDB
    /// </summary>
    /// <param name="senderID">The DE SenderID</param>
    /// <param name="distributionID">The DE DistributionID</param>
    private void DeleteDEFromCache(string senderID, string distributionID)
    {
      QueryRequest itemstodelete = new QueryRequest();
      itemstodelete.TableName = AWSConstants.DEProcessing.DynamoInfo.TableName;
      itemstodelete.KeyConditions = new Dictionary<string, Condition>();
      Condition hashcond = new Condition();
      hashcond.ComparisonOperator = "EQ";
      hashcond.AttributeValueList = new List<AttributeValue>();
      AttributeValue hashvalue = new AttributeValue();
      hashvalue.S = senderID + ">" + distributionID;
      hashcond.AttributeValueList.Add(hashvalue);
      itemstodelete.KeyConditions.Add("DEID", hashcond);
      QueryResponse itemstodeleteresp = this.amazonDynamoClient.Query(itemstodelete);
      List<Dictionary<string, AttributeValue>> result = itemstodeleteresp.Items;
      foreach (Dictionary<string, AttributeValue> dbitem in result)
      {
        DeleteItemRequest delreq = new DeleteItemRequest();
        delreq.TableName = AWSConstants.DEProcessing.DynamoInfo.TableName;
        AttributeValue key = new AttributeValue();
        key.S = dbitem["DEID"].S;
        delreq.Key.Add("DEID", key);
        key = new AttributeValue();
        key.S = dbitem["DistributionTime"].S;
        delreq.Key.Add("DistributionTime", key);
        try
        {
          DeleteItemResponse delresp = this.amazonDynamoClient.DeleteItem(delreq);
          if (delresp.HttpStatusCode == System.Net.HttpStatusCode.OK)
          {
            Log.Info("Item deleted from dynamo");
          }
          else
          {
            Log.Error("Error deleting from dynamo" + delresp.HttpStatusCode.ToString());
          }
        }
        catch (Exception e)
        {
          Log.Error("Error deleting from dynamo", e);
        }
      }
    }

    /// <summary>
    /// Queries the database to see if the specified message is the latest message
    /// </summary>
    /// <param name="senderID">SenderID of the DE Message</param>
    /// <param name="distributionID">DistributionID of the DE Message</param>
    /// <param name="distributionTime">DistributionTime of the DE Message</param>
    /// <returns>True if the message is the latest (e.g. ok to delete other messages)</returns>
    private bool IsLatestDE(string senderID, string distributionID, DateTime distributionTime)
    {
      QueryRequest latestquery = new QueryRequest();
      latestquery.TableName = AWSConstants.DEProcessing.DynamoInfo.TableName;
      latestquery.KeyConditions = new Dictionary<string, Condition>();

      Condition hashcond = new Condition();
      hashcond.ComparisonOperator = "EQ";
      hashcond.AttributeValueList = new List<AttributeValue>();
      AttributeValue hashvalue = new AttributeValue();
      hashvalue.S = distributionID;
      hashcond.AttributeValueList.Add(hashvalue);
      latestquery.KeyConditions.Add("distributionID", hashcond);

      hashcond = new Condition();
      hashcond.ComparisonOperator = "EQ";
      hashcond.AttributeValueList = new List<AttributeValue>();
      hashvalue = new AttributeValue();
      hashvalue.S = senderID;
      hashcond.AttributeValueList.Add(hashvalue);
      latestquery.KeyConditions.Add("senderID", hashcond);

      try
      {
        QueryResponse itemstodeleteresp = this.amazonDynamoClient.Query(latestquery);
        if (itemstodeleteresp.Items.Count == 0)
        {
          Log.Debug("No existing DE in cache.");
          return true;
        }
        else
        {
          DateTime foundSentTime = DateTime.ParseExact(itemstodeleteresp.Items[0]["dateTimeSent"].S, AWSSDKUtils.ISO8601DateFormat, new DateTimeFormatInfo());
          Log.DebugFormat("Comparing new {0} >? {1}", distributionTime, foundSentTime);
          return distributionTime > foundSentTime;
        }
      }
      catch (Exception e)
      {
        Log.Error("Error querying from dynamo for latest", e);
        return false;
      }
    }
  }
}
